package com.hhf.rrd.usage;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.ExplainDetail;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.function.Consumer;

/**
 * table sql 时间用法
 *      -- 待验证
 *
 * @author huanghaifeng15
 * @date 2022/2/15 16:44
 **/
public class TableUsageApp {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 指定一个watermark
        String sql5 = "CREATE TABLE user_actions (\n" +
                "  user_name STRING,\n" +
                "  data STRING,\n" +
                "  user_action_time TIMESTAMP(3),\n" +
                "  -- 使用下面这句来将 user_action_time 声明为事件时间，并且声明 watermark 的生成规则，即 user_action_time 减 5 秒\n" +
                "  -- 事件时间列的字段类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型\n" +
                "  WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen'\n" +
                ")\n";
        tEnv.executeSql(sql5);

        String sql6 = "SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)\n" +
                "FROM user_actions\n" +
                "-- 然后就可以在窗口算子中使用 user_action_time\n" +
                "GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE)\n";
        tEnv.executeSql(sql6).print();

        // 创建一个输出表
        String sql4 =
                "CREATE TABLE sink_table2 (\n" +
                        "  c1 TIMESTAMP,\n" +
                        "  c2 TIMESTAMP,\n" +
                        "  c3 BIGINT,\n" +
                        "  c4 STRING,\n" +
                        "  c5 VARCHAR,\n" +
                        "  c6 VARCHAR,\n" +
                        "  c7 BIGINT,\n" +
                        "  c8 TIMESTAMP,\n" +
                        "  c9 TIMESTAMP,\n" +
                        "  c10 TIMESTAMP,\n" +
                        "  c11 BIGINT\n" +
                        ") WITH (\n" +
                        "  'connector' = 'print'\n" +
                        ")";
        tEnv.executeSql(sql4);
        tEnv.executeSql("DESCRIBE sink_table2").print();

        // xxx
        // String sql4 = "";
        // tEnv.executeSql(sql4);

        // 时间转换
        String sql3 = "INSERT INTO sink_table2\n" +
                "SELECT \n" +
                "TO_TIMESTAMP('2020-08-03 10:59:45.957') as c1, \n" +
                "TIMESTAMP '2020-08-03 10:59:45.957' as c2, \n" +
                "UNIX_TIMESTAMP('2020-08-03 10:59:45.957') as c3, \n" +
                "DATE_FORMAT(FROM_UNIXTIME(1596702949000), 'yyyy-MM-dd HH:mm:ss') as c4, \n" +
                "FROM_UNIXTIME(1596702949000/1000) as c5, \n" +
                "FROM_UNIXTIME(1596702949000/1000) as c6, \n" +
                "EXTRACT(DAY FROM TO_TIMESTAMP('2020-08-03 10:59:45.957')) as c7, -- 20:东八区的20日\n" +
                "LOCALTIMESTAMP as c8,\n" +
                "CURRENT_TIMESTAMP as c9,\n" +
                "NOW() as c10,\n" +
                "UNIX_TIMESTAMP() as c11\n";
        tEnv.executeSql(sql3).print();
        tEnv.explainSql(sql3,ExplainDetail.ESTIMATED_COST,ExplainDetail.ESTIMATED_COST);

        // 创建一个输出表
        String sql2 =
                "CREATE TABLE sink_table (\n" +
                "    result_interval_year TIMESTAMP(3),\n" +
                "    result_interval_year_p TIMESTAMP(3),\n" +
                "    result_interval_year_p_to_month TIMESTAMP(3),\n" +
                "    result_interval_month TIMESTAMP(3),\n" +
                "    result_interval_day TIMESTAMP(3),\n" +
                "    result_interval_day_p1 TIMESTAMP(3),\n" +
                "    result_interval_day_p1_to_hour TIMESTAMP(3),\n" +
                "    result_interval_day_p1_to_minute TIMESTAMP(3),\n" +
                "    result_interval_day_p1_to_second_p2 TIMESTAMP(3),\n" +
                "    result_interval_hour TIMESTAMP(3),\n" +
                "    result_interval_hour_to_minute TIMESTAMP(3),\n" +
                "    result_interval_hour_to_second TIMESTAMP(3),\n" +
                "    result_interval_minute TIMESTAMP(3),\n" +
                "    result_interval_minute_to_second_p2 TIMESTAMP(3),\n" +
                "    result_interval_second TIMESTAMP(3),\n" +
                "    result_interval_second_p2 TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'print'\n" +
                ")";
        tEnv.executeSql(sql2).print();

        // INTERVAL 间隔
        String sql1 = "INSERT INTO sink_table\n" +
                "SELECT\n" +
                "    -- Flink SQL 支持的所有 INTERVAL 子句如下，总体可以分为 `年-月`、`日-小时-秒` 两种\n" +
                "    -- 1. 年-月。取值范围为 [-9999-11, +9999-11]，其中 p 是指有效位数，取值范围 [1, 4]，默认值为 2。比如如果值为 1000，但是 p = 2，则会直接报错。\n" +
                "    -- INTERVAL YEAR\n" +
                "    TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10' YEAR as result_interval_year\n" +
                "    -- INTERVAL YEAR(p)\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '100' YEAR(3) as result_interval_year_p\n" +
                "    -- INTERVAL YEAR(p) TO MONTH\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10-03' YEAR(3) TO MONTH as result_interval_year_p_to_month\n" +
                "    -- INTERVAL MONTH\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '13' MONTH as result_interval_month\n" +
                "    -- 2. 日-小时-秒。取值范围为 [-999999 23:59:59.999999999, +999999 23:59:59.999999999]，其中 p1\\p2 都是有效位数，p1 取值范围 [1, 6]，默认值为 2；p2 取值范围 [0, 9]，默认值为 6\n" +
                "    -- INTERVAL DAY\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10' DAY as result_interval_day\n" +
                "    -- INTERVAL DAY(p1)\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '100' DAY(3) as result_interval_day_p1\n" +
                "    -- INTERVAL DAY(p1) TO HOUR\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10 03' DAY(3) TO HOUR as result_interval_day_p1_to_hour\n" +
                "    -- INTERVAL DAY(p1) TO MINUTE\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10 03:12' DAY(3) TO MINUTE as result_interval_day_p1_to_minute\n" +
                "    -- INTERVAL DAY(p1) TO SECOND(p2)\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10 00:00:00.004' DAY TO SECOND(3) as result_interval_day_p1_to_second_p2\n" +
                "    -- INTERVAL HOUR\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10' HOUR as result_interval_hour\n" +
                "    -- INTERVAL HOUR TO MINUTE\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10:03' HOUR TO MINUTE as result_interval_hour_to_minute\n" +
                "    -- INTERVAL HOUR TO SECOND(p2)\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '00:00:00.004' HOUR TO SECOND(3) as result_interval_hour_to_second\n" +
                "    -- INTERVAL MINUTE\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '10' MINUTE as result_interval_minute\n" +
                "    -- INTERVAL MINUTE TO SECOND(p2)\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '05:05.006' MINUTE TO SECOND(3) as result_interval_minute_to_second_p2\n" +
                "    -- INTERVAL SECOND\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '3' SECOND as result_interval_second\n" +
                "    -- INTERVAL SECOND(p2)\n" +
                "    , TO_TIMESTAMP('1970-01-01 08:00:00.000') + INTERVAL '300' SECOND(3) as result_interval_second_p2";
        tEnv.executeSql(sql1).print();
    }

}
